package server;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.DatagramChannel;
import java.util.HashSet;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;

import server.Message.MsgType;

public class UserSendThread extends Thread {

	/* Map pubIds -> subscriber group */
	private ConcurrentHashMap<String, HashSet<String>> subscribers;

	private Map<String, CopyOnWriteArrayList<SocketAddress>> sAddrs;
	
	private ConcurrentHashMap<String, SocketAddress> uAddrs;
	private BlockingQueue<Message> outgoing;

	private InetSocketAddress liveAddr;
	private DatagramChannel sock;
	private ByteBuffer out;

	public UserSendThread(BlockingQueue<Message> outgoing,
			ConcurrentHashMap<String, SocketAddress> uAddrs,
			DatagramChannel uSock, Map<String, CopyOnWriteArrayList<SocketAddress>> sAddrs, int livePort) {
		super("UserSend-" + uSock.socket().getLocalPort());
		this.outgoing = outgoing;
		this.uAddrs = uAddrs;
		this.sAddrs = sAddrs;

		this.liveAddr = new InetSocketAddress("localhost", livePort);
		this.sock = uSock;

		this.subscribers = new ConcurrentHashMap<String, HashSet<String>>();
	}

	private void log(String msg) {
		Server.log(this.getName(), msg);
	}

	@Override
	public void run() {
		while (true) {
			try {
				Message m = outgoing.take();
				switch (m.type) {
				case SURVEY:
					notifyLiveness(m.pubId, m.data, m.flags);
					break;
				case CONNECT:
					SocketAddress to = m.addr;
					if ((m.flags & 0x2) > 0) {
						uAddrs.put(m.pubId, to);
						
						if ((m.flags & 0x1) == 0) {
							if (!sAddrs.containsKey(m.data)) {
								sAddrs.put(m.data, Server.initMesh());
								notifyLiveness(m.data);
							}
							HashSet<String> subs;
							if (!subscribers.containsKey(m.data)) {
								subs = new HashSet<String>();
								subscribers.put(m.data, subs);
							} else {
								subs = subscribers.get(m.data);
							}
							subs.add(m.pubId);
							log("Subscriber : " + m.pubId);
							log("Subscribers: " + subs.size());
							log("Connected  : " + uAddrs.size());
						}
					}
					
					out = m.serialize();
					try {
						log("Sending... " + m);
						sock.send(out, to);
						out.flip();
					} catch (IOException e) {
						e.printStackTrace();
					}
					break;
				case SUB:
					HashSet<String> subs;
					if (!subscribers.containsKey(m.data)) {
						subs = new HashSet<String>();
						subscribers.put(m.data, subs);
					} else {
						subs = subscribers.get(m.data);
					}
					subs.add(m.pubId);
					log("Subscriber : " + m.pubId);
					log("Subscribers: " + subs.size());
					log("Connected  : " + uAddrs.size());
					try {
						out = m.serialize();
						sock.send(out, uAddrs.get(m.pubId));
						out.flip();
					} catch (IOException e) {
						e.printStackTrace();
					}
					break;
				case PUB:
				default:
					HashSet<String> receivers = subscribers.get(m.pubId);
					if (receivers != null) {
						out = m.serialize();
						for (String s : receivers) {
							try {
								sock.send(out, uAddrs.get(s));
								out.flip();
							} catch (IOException e) {
								e.printStackTrace();
							}
						}
					}
				}
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
	}
	
	private void notifyLiveness(String mesh, String id, byte flags) {
		log("Notifying liveness to survey '" + mesh + "'");
		try {
			String[] args = id.split(" ");
			Message notify;
			if (args.length == 1)
				notify = new Message(MsgType.SURVEY, mesh, 0, args[0] + " " + Server.calculateLoad(uAddrs.size()));
			else
				notify = new Message(MsgType.SURVEY, mesh, 0, args[0] + " " + Server.calculateLoad(uAddrs.size()) + " " + args[1]); 
			notify.flags = flags;
			sock.send(notify.serialize(), liveAddr);
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}
	
	private void notifyLiveness(String id) {
		log("Notifying liveness to join '" + id + "'");
		try {
			Message notify = new Message(MsgType.REQUEST, id, 0, "");
			sock.send(notify.serialize(), liveAddr);
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}
}
